package redis.client.nedis;

import lombok.SneakyThrows;
import redis.client.nedis.request.Protocol;
import redis.client.nedis.request.RequestPool;

import java.util.concurrent.*;

/**
 * redis结果解析
 *
 * @author gaodong
 */
public class RedisResultDeserializer extends Thread {

    /**
     * 服务器地址
     */
    private HostAndPort hostAndPort;

    /**
     * 接收的字字节流
     */
    private BlockingQueue<Byte> revByteQueue = new LinkedBlockingQueue<Byte>();

    /**
     * 信号量，用于通知可以继续解析
     */
    private Semaphore mutex = new Semaphore(1);

    public RedisResultDeserializer(HostAndPort hp) {
        this.hostAndPort = hp;
        this.setDaemon(true);
    }

    /**
     * 添加到队列
     *
     * @param b
     */
    public void offer(Byte b) {
        revByteQueue.offer(b);
    }

    /**
     * 添加到队列
     */
    public void notifyRelease() {
        mutex.release();
    }

    @SneakyThrows
    @Override
    public void run() {

        do {
            Byte b = revByteQueue.take();
            byte bv = b.byteValue();
            switch (bv) {
                // 错误回复（error reply）的第一个字节是 "-"
                case Protocol.ERROR_REPLAY:
                    String reply = readLine();
                    //RequestPool.putResult(hostAndPort, reply);
                    break;

                // 状态回复（status reply）的第一个字节是 "+"
                case Protocol.STATUS_REPLAY:
                    reply = readLine();
                    //RequestPool.putResult(hostAndPort, reply);
                    break;

                // 整数回复（integer reply）的第一个字节是 ":"
                case Protocol.INTEGER_REPLAY:
                    Long v = readLong();
                    //RequestPool.putResult(hostAndPort, v);
                    break;

                // 多条批量回复（multi bulk reply）的第一个字节是 "*"
                case Protocol.MULTI_BULK_REPLAY:
                    break;

                // 批量回复（bulk reply）的第一个字节是 "$"
                case Protocol.BULK_REPLAY:
                    long len = readLong();
                    if (len == -1) {
                        reply = null;
                    } else {
                        reply = readLine(len);
                    }
                    //RequestPool.putResult(hostAndPort, reply);
                    break;
            }

        } while (Thread.currentThread().isAlive());
    }

    /**
     * 获取下一个byte字符
     *
     * @return
     */
    private byte getNextByte() throws InterruptedException {
        return revByteQueue.take().byteValue();
    }

    /**
     * 读取一行String
     *
     * @return
     */
    private String readLine() throws InterruptedException {

        StringBuilder sb = new StringBuilder();
        do {
            byte b = getNextByte();
            while (b == Protocol.CR) {
                b = getNextByte();
                if (b == Protocol.LF) {
                    String reply = sb.toString();
                    return reply;
                }
            }
            sb.append((char) b);
        } while (Thread.currentThread().isAlive());

        return null;
    }

    /**
     * 读取一行String，指定长度
     *
     * @return
     */
    private String readLine(long len) throws InterruptedException {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < len; i++) {
            sb.append((char) getNextByte());
        }
        skipCRLF();
        return sb.toString();
    }

    /**
     * 跳过\r\n
     */
    private void skipCRLF() throws InterruptedException {
        do {
            byte b = getNextByte();
            while (b == Protocol.CR) {
                b = getNextByte();
                if (b == Protocol.LF) {
                    return;
                }
            }
        } while (Thread.currentThread().isAlive());
    }

    /**
     * 读取一行String
     *
     * @return
     */
    private Long readLong() throws InterruptedException {

        boolean isNeg = false;
        long v = 0;
        do {
            byte b = getNextByte();

            if (b == '-') {
                isNeg = true;
                b = getNextByte();
            }

            while (b == Protocol.CR) {
                b = getNextByte();
                if (b == Protocol.LF) {
                    return isNeg ? -v : v;
                }
            }
            v = 10 * v + b - '0';

        } while (Thread.currentThread().isAlive());

        return null;
    }
}
